package cn.itcast.flink.base;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * Author itcast
 * Date 2021/7/27 16:45
 * Desc TODO
 */
public class MySQLSink {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source
        DataStream<Student> studentDS = env.fromElements(new Student(null, "xiaohua", 20));
        //3.Transformation
        //4.Sink
        //4.Sink 实现自定义 MySQL sink
        DataStreamSink<Student> source = studentDS.addSink(new MyStudent());
        //5.execute
        env.execute();
        //创建 Student 类，包含3个字段 id name age
        //创建 MySQLSink 类继承 RichSinkFunction<Student>
        //实现 open invoke close 方法
    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }

    public static class MyStudent extends RichSinkFunction<Student>{
        private Connection conn = null;
        private PreparedStatement ps = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            //加载驱动,开启连接
            //Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://node3:3306/bigdata?useSSL=false", "root", "123456");
            String sql = "INSERT INTO `t_student` (`id`, `name`, `age`) VALUES (null, ?, ?)";
            ps = conn.prepareStatement(sql);
        }

        @Override
        public void invoke(Student value, Context context) throws Exception {
            ps.setString(1,value.name);
            ps.setInt(2,value.age);
            ps.executeUpdate();
        }

        @Override
        public void close() throws Exception {
            super.close();
            if (conn != null) conn.close();
            if (ps != null) ps.close();
        }
    }

}
